package com.wl.mr.join;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

public class DeliveReducer extends Reducer<Text,DeliveBean,DeliveBean, NullWritable> {
    @Override
    protected void reduce(Text key, Iterable<DeliveBean> values, Context context) throws IOException, InterruptedException {

        DeliveBean positionBean = new DeliveBean();
        ArrayList<DeliveBean> deBean = new ArrayList<>();
        for (DeliveBean bean : values) {
            if(bean.getFlag().equalsIgnoreCase("deliver")){
                DeliveBean newBean = new DeliveBean();
                try {
                    BeanUtils.copyProperties(newBean,bean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
                deBean.add(newBean);

            }else {
                try {
                    BeanUtils.copyProperties(positionBean,bean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
            }
        }

        for (DeliveBean bean : deBean) {
            bean.setPositionName(positionBean.getPositionName());
            context.write(bean,NullWritable.get());
        }

    }
}
